<?php

namespace app\framework;

use app\Console;
use app\framework\kline\CryptoCurrencyKline;
use app\framework\lib\Mysql;
use app\framework\lib\Redis; // 假设已实现Redis单例工具类
use app\framework\lib\Config;
use app\framework\lib\Dict;
use Workerman\Timer;
use Workerman\Connection\TcpConnection;
use Workerman\Worker;
use Throwable;
use app\framework\lib\Log;
class Server
{
    public $worker = null;

    // Redis键名常量（全局共享连接和订阅信息）
    private const REDIS_CONN_SUB_KEY = 'server:connection:sub:%s'; // 存储连接的订阅信息（%s替换为connection->id）

    /**
     * 服务入口
     * @param array $config
     */
    public static function run(array $config): void
    {
        $server = new self($config);
        $server->start();
    }

    public function __construct(array $config)
    {
        Config::init($config);
    }

    public function start(): void
    {
        $serverConfig = Config::getInstance()->server;
        // 验证核心配置
        $this->validateConfig($serverConfig);

        $worker = new Worker($serverConfig["proto"]);
        $worker->count = (int)$serverConfig["count"];
        $worker->name = $serverConfig["name"];
        Worker::$logFile = $serverConfig["log_file"];
        Worker::$stdoutFile = $serverConfig["stdout_file"];
        Worker::$pidFile = $serverConfig["pid_file"];

        // 绑定回调（职责单一化）
        $worker->onWorkerStart = [$this, 'onWorkerStart'];
        $worker->onWorkerStop = [$this, 'onWorkerStop'];
        $worker->onConnect = [$this, 'onConnect'];
        $worker->onMessage = [$this, 'onMessage'];
        $worker->onClose = [$this, 'onClose'];

        $this->worker = $worker;
        Worker::runAll();
    }

    /**
     * 进程启动回调（初始化资源+绑定定时器）
     * @param Worker $worker
     */
    public function onWorkerStart(Worker $worker): void
    {
        try {
            // 初始化进程内资源（每个进程独立初始化）
            Mysql::start($worker);
            Dict::getInstance()->init();
            // 进程分工（按ID执行不同任务）
           $this->initWorkerTasks($worker);

            // 所有进程都启动定时器（处理当前进程的连接）
           $this->startConnectionTimer($worker);
        } catch (Throwable $e) {
            Log::error("[进程{$worker->id}启动失败",["message"=>$e->getMessage(),"file"=>$e->getFile(),"trace"=>$e->getTrace()]);
            // 启动失败则退出进程
            Worker::stopAll();
        }
    }

    /**
     * 进程停止回调（释放资源）
     * @param Worker $worker
     */
    public function onWorkerStop(Worker $worker): void
    {

        echo "[进程{$worker->id}停止] 资源已释放\n";
    }

    /**
     * 客户端连接回调（初始化订阅信息）
     * @param TcpConnection $connection
     */
    public function onConnect(TcpConnection $connection): void
    {
        $connId = $connection->id;
        $workerId = $this->worker->id;

        try {
            // 用Redis存储连接的订阅信息（多进程共享）
            Redis::getInstance()->set(
                sprintf(self::REDIS_CONN_SUB_KEY, $connId),
                json_encode([]) // 初始化为空订阅
            );
            echo "[进程{$workerId}] 新连接：{$connId}（已初始化订阅）\n";
        } catch (Throwable $e) {
            echo "[进程{$workerId}] 连接{$connId}初始化失败：{$e->getMessage()}\n";
            $connection->close(); // 初始化失败则关闭连接
        }
    }

    /**
     * 客户端断开回调（清理订阅信息）
     * @param TcpConnection $connection
     */
    public function onClose(TcpConnection $connection): void
    {
        $connId = $connection->id;
        $workerId = $this->worker->id;

        try {
            // 从Redis删除订阅信息（避免内存泄漏）
            Redis::getInstance()->del(sprintf(self::REDIS_CONN_SUB_KEY, $connId));
            echo "[进程{$workerId}] 连接{$connId}断开：已清理订阅\n";
        } catch (Throwable $e) {
            echo "[进程{$workerId}] 连接{$connId}清理失败：{$e->getMessage()}\n";
        }
    }

    /**
     * 处理客户端消息（更新订阅）
     * @param TcpConnection $connection
     * @param mixed $data
     */
    public function onMessage(TcpConnection $connection, $data): void
    {
        $connId = $connection->id;
        $workerId = $this->worker->id;

        try {
            $data = json_decode($data, true);
            // 验证消息格式
            if (json_last_error() !== JSON_ERROR_NONE || !is_array($data)) {
                throw new \InvalidArgumentException("无效的消息格式");
            }

            switch ($data["type"] ?? '') {
                case "init":
                    // 处理订阅（去重+存储到Redis）
                    $subCodes = isset($data["data"]) && is_array($data["data"])
                        ? array_unique($data["data"])
                        : [];
                    Redis::getInstance()->set(
                        sprintf(self::REDIS_CONN_SUB_KEY, $connId),
                        json_encode($subCodes)
                    );
                    $connection->send(json_encode(["type" => "pong", "msg" => "订阅成功"]));
                    echo "[进程{$workerId}] 连接{$connId}订阅：" . implode(',', $subCodes) . "\n";
                    break;
                case 'ping':

                    break;
                default:
                    $connection->send(json_encode(["type" => "error", "msg" => "不支持的消息类型"]));
            }
        } catch (Throwable $e) {
            $errorMsg = "[进程{$workerId}] 连接{$connId}消息处理失败：{$e->getMessage()}";
            echo $errorMsg . "\n";
            $connection->send(json_encode(["type" => "error", "msg" => $errorMsg]));
        }
    }

    /**
     * 验证服务配置合法性
     * @param array $config
     * @throws \InvalidArgumentException
     */
    private function validateConfig(array $config): void
    {
        $required = ["proto", "count", "name", "log_file", "stdout_file", "pid_file"];
        foreach ($required as $key) {
            if (empty($config[$key])) {
                throw new \InvalidArgumentException("服务配置缺失：{$key}");
            }
        }
        if ((int)$config["count"] < 1) {
            throw new \InvalidArgumentException("进程数count必须≥1");
        }
    }

    /**
     * 初始化不同进程的任务（按ID分工）
     * @param Worker $worker
     */
    private function initWorkerTasks(Worker $worker): void
    {
        switch ($worker->id) {
            case 1:
                CryptoCurrencyKline::getInstance()->init($worker);
                Console::command();
                Log::info("[进程1] 已初始化Kline任务和command");
                break;
            case 0:
                $serverConfig = Config::getInstance()->server;
                Console::currency($serverConfig, $worker);
                Log::info("[进程0] 已初始化currency任务");
                break;
            // 其他进程可扩展更多任务
        }
    }

    /**
     * 启动连接处理定时器（每个进程处理自己的连接）
     * @param Worker $worker
     */
    private function startConnectionTimer(Worker $worker): void
    {
        Timer::add(2, function () use ($worker) {
            try {
                $workerId = $worker->id;
                foreach ($worker->connections as $connection) {

                    $this->processConnection($connection, $workerId);
                }
            } catch (Throwable $e) {
                Log::error("[进程{$worker->id}定时器错误]",[
                    "error"=>$e->getTrace(),
                    "msg"=>$e->getMessage(),
                    "file"=>$e->getFile(),
                    "line"=>$e->getLine()
                ]);
            }
        });
        Timer::add(3, function () use ($worker) {
            try {

                foreach ($worker->connections as $connection) {

                   $connection->send(json_encode(["type"=>"ping"]));
                }
            } catch (Throwable $e) {
                Log::error("[进程{$worker->id}定时器错误]",[
                    "error"=>$e->getTrace(),
                    "msg"=>$e->getMessage(),
                    "file"=>$e->getFile(),
                    "line"=>$e->getLine()
                ]);
            }
        });
    }

    /**
     * 处理单个连接（获取订阅+推送数据）
     * @param TcpConnection $connection
     * @param int $workerId
     */
    private function processConnection(TcpConnection $connection, int $workerId): void
    {
        $connId = $connection->id;

        // 从Redis获取订阅信息（多进程共享）
        $subJson = Redis::getInstance()->get(sprintf(self::REDIS_CONN_SUB_KEY, $connId));
        $subscribedCodes = json_decode($subJson, true) ?? [];
        if (empty($subscribedCodes)) {
            echo "[进程{$workerId}] 连接{$connId}无订阅信息，跳过\n";
            return;
        }
        // 获取行情数据（验证有效性）
        $marketData = Console::getData($subscribedCodes);
        if (!is_array($marketData)) {
            echo "[进程{$workerId}] 连接{$connId}获取行情数据失败\n";
            return;
        }

        // 推送数据
        $connection->send(json_encode([
            'type' => 'market',
            'time' => date('Y-m-d H:i:s'),
            'data' => $marketData
        ]));
    }
}